from fastapi import APIRouter, Depends, HTTPException, status, Query, Path, BackgroundTasks, Form
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.sql import select, func, and_
from typing import List, Optional, Dict, Any
from datetime import datetime
import json
import logging
from app.datatk.data_help import DataHelp
from app.models.db import DataSource, DataSet, DataField, FieldChangeLog, User, UserMsg, UserSession, get_async_session
from app.models.user import current_active_user, fastapi_users

from app.services.dataset_service import (
    check_structure_changes,
    apply_structure_changes,
    sync_dataset,
    update_field_metadata,
    get_field_relationship,
    generate_metadata_suggestions
)
from app.services.datasource_service import sync_datasource_fields_to_dataset

router = APIRouter(
    prefix="/datasets",
    tags=["数据集管理"]
)


# ==================== 数据集管理 ====================

@router.post("/", response_model=Dict[str, Any])
async def create_dataset(
    dataset_data: Dict[str, Any],
    current_user: User = Depends(current_active_user),
    session: AsyncSession = Depends(get_async_session)
):
    """创建数据集
    
    Args:
        dataset_data: 数据集数据
    """
    try:
        # 检查数据源是否存在
        data_source_id = dataset_data["data_source_id"]
        result = await session.execute(
            select(DataSource).filter(DataSource.id == data_source_id)
        )
        data_source = result.scalar_one_or_none()
        
        if not data_source:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"数据源不存在: {data_source_id}"
            )
        
        # 创建数据集对象
        dataset = DataSet(
            data_source_id=data_source_id,
            name=dataset_data["name"],
            description=dataset_data.get("description"),
            original_name=dataset_data["original_name"],
            type=dataset_data["type"],
            custom_query=dataset_data.get("custom_query"),
            category=dataset_data.get("category"),
            sync_enabled=dataset_data.get("sync_enabled", True),
            sync_frequency=dataset_data.get("sync_frequency", "daily"),
            sync_type=dataset_data.get("sync_type", "full"),
            incremental_field=dataset_data.get("incremental_field"),
            structure_change_strategy=dataset_data.get("structure_change_strategy", "notify"),
            created_at=datetime.now(),
            sync_status="pending",
            user_id=str(current_user.id)
        )
        
        # 设置标签列表
        if "tags" in dataset_data:
            dataset.set_tags_list(dataset_data["tags"])
        
        # 添加到数据库
        session.add(dataset)
        await session.commit()
        await session.refresh(dataset)
        
        return {
            "success": True,
            "message": "数据集创建成功",
            "data": dataset.to_dict()
        }
    
    except HTTPException:
        raise
    
    except Exception as e:
        await session.rollback()
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"创建数据集失败: {str(e)}"
        )


@router.get("/", response_model=Dict[str, Any])
async def get_datasets(
    skip: int = Query(0, ge=0),
    limit: int = Query(100, ge=1, le=1000),
    search: Optional[str] = None,
    data_source_id: Optional[str] = None,
    category: Optional[str] = None,
    tags: Optional[str] = None,
    include_excel: bool = Query(False, description="是否包含Excel文件上传的数据集"),
    session: AsyncSession = Depends(get_async_session)
):
    """获取数据集列表
    
    Args:
        skip: 跳过记录数
        limit: 返回记录数上限
        search: 搜索关键词
        data_source_id: 数据源ID
        category: 分类
        tags: 标签（逗号分隔）
        include_excel: 是否包含用户上传的Excel文件数据集（默认为False）
    """
    try:
        # 构建查询
        query = select(DataSet).order_by(DataSet.created_at.desc())
        
        # 应用过滤条件
        if search:
            query = query.filter(
                (DataSet.name.ilike(f"%{search}%")) | 
                (DataSet.description.ilike(f"%{search}%")) | 
                (DataSet.original_name.ilike(f"%{search}%"))
            )
        
        if data_source_id:
            query = query.filter(DataSet.data_source_id == data_source_id)
        
        # 排除用户上传的Excel文件数据集（数据源ID为"00000000-0000-0000-0000-000000000000"）
        if not include_excel:
            query = query.filter(DataSet.data_source_id != '00000000-0000-0000-0000-000000000000')
        
        if category:
            query = query.filter(DataSet.category == category)
        
        # 标签过滤（简化处理）
        if tags:
            tag_list = tags.split(",")
            for tag in tag_list:
                query = query.filter(DataSet.tags.ilike(f"%{tag.strip()}%"))
        
        # 计算总数 - 使用子查询方式
        total_query = select(func.count()).select_from(query.subquery())
        total_count = await session.scalar(total_query)
        
        # 执行查询
        result = await session.execute(query.offset(skip).limit(limit))
        datasets = result.scalars().all()
        
        # 构建返回结果
        return {
            "success": True,
            "data": [ds.to_dict() for ds in datasets],
            "total": total_count,
            "skip": skip,
            "limit": limit
        }
    
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"获取数据集列表失败: {str(e)}"
        )


@router.get("/{dataset_id}", response_model=Dict[str, Any])
async def get_dataset(
    dataset_id: str = Path(..., title="数据集ID"),
    current_user: User = Depends(current_active_user),
    session: AsyncSession = Depends(get_async_session)
):
    """获取数据集详情
    
    Args:
        dataset_id: 数据集ID
    """
    try:
        # 查询数据集
        result = await session.execute(
            select(DataSet).filter(DataSet.id == dataset_id)
        )
        dataset = result.scalar_one_or_none()
        
        if not dataset:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"数据集不存在: {dataset_id}"
            )
        
        # 获取数据源信息
        datasource_result = await session.execute(
            select(DataSource).filter(DataSource.id == dataset.data_source_id)
        )
        datasource = datasource_result.scalar_one_or_none()
        
        # 获取字段信息
        fields_result = await session.execute(
            select(DataField).filter(
                DataField.dataset_id == dataset_id
            ).order_by(DataField.ordinal_position)
        )
        fields = fields_result.scalars().all()
        
        # 构建数据集详细信息
        dataset_dict = dataset.to_dict()
        dataset_dict["data_source"] = datasource.to_dict() if datasource else None
        dataset_dict["fields"] = [field.to_dict() for field in fields]
        
        return {
            "success": True,
            "data": dataset_dict
        }
    
    except HTTPException:
        raise
    
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"获取数据集详情失败: {str(e)}"
        )


@router.put("/{dataset_id}", response_model=Dict[str, Any])
async def update_dataset(
    dataset_data: Dict[str, Any],
    dataset_id: str = Path(..., title="数据集ID"),
    current_user: User = Depends(current_active_user),
    session: AsyncSession = Depends(get_async_session)
):
    """更新数据集
    
    Args:
        dataset_id: 数据集ID
        dataset_data: 更新数据
    """
    try:
        # 查询数据集
        result = await session.execute(
            select(DataSet).filter(DataSet.id == dataset_id)
        )
        dataset = result.scalar_one_or_none()
        
        if not dataset:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"数据集不存在: {dataset_id}"
            )

        
        # 更新基本信息
        if "name" in dataset_data:
            dataset.name = dataset_data["name"]
        
        if "description" in dataset_data:
            dataset.description = dataset_data["description"]
        
        if "category" in dataset_data:
            dataset.category = dataset_data["category"]
        
        if "tags" in dataset_data:
            dataset.set_tags_list(dataset_data["tags"])
        
        # 更新同步配置
        if "sync_enabled" in dataset_data:
            dataset.sync_enabled = dataset_data["sync_enabled"]
        
        if "sync_frequency" in dataset_data:
            dataset.sync_frequency = dataset_data["sync_frequency"]
        
        if "sync_type" in dataset_data:
            dataset.sync_type = dataset_data["sync_type"]
        
        if "incremental_field" in dataset_data:
            dataset.incremental_field = dataset_data["incremental_field"]
        
        if "structure_change_strategy" in dataset_data:
            dataset.structure_change_strategy = dataset_data["structure_change_strategy"]
        
        # 更新自定义查询
        if "custom_query" in dataset_data:
            dataset.custom_query = dataset_data["custom_query"]
        
        # 更新时间
        dataset.updated_at = datetime.now()
        
        # 保存到数据库
        session.add(dataset)
        await session.commit()
        await session.refresh(dataset)
        
        return {
            "success": True,
            "message": "数据集更新成功",
            "data": dataset.to_dict()
        }
    
    except HTTPException:
        raise
    
    except Exception as e:
        await session.rollback()
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"更新数据集失败: {str(e)}"
        )


@router.delete("/{dataset_id}", response_model=Dict[str, Any])
async def delete_dataset(
    dataset_id: str = Path(..., title="数据集ID"),
    current_user: User = Depends(current_active_user),
    session: AsyncSession = Depends(get_async_session)
):
    """删除数据集
    
    Args:
        dataset_id: 数据集ID
    """
    try:
        # 查询数据集
        result = await session.execute(
            select(DataSet).filter(DataSet.id == dataset_id)
        )
        dataset = result.scalar_one_or_none()
        
        if not dataset:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"数据集不存在: {dataset_id}"
            )
        
        # 先删除与此数据集相关的角色数据集访问权限记录
        from sqlalchemy import text
        # 使用原始SQL删除关联记录，避免ORM的外键约束问题
        await session.execute(
            text("DELETE FROM dk_role_dataset_access WHERE dataset_id = :dataset_id"),
            {"dataset_id": dataset_id}
        )
        
        # 删除数据集（级联删除会自动删除关联的字段和变更日志）
        await session.delete(dataset)
        await session.commit()
        
        return {
            "success": True,
            "message": "数据集删除成功"
        }
    
    except HTTPException:
        raise
    
    except Exception as e:
        await session.rollback()
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"删除数据集失败: {str(e)}"
        )


@router.post("/{dataset_id}/check-structure", response_model=Dict[str, Any])
async def check_structure(
    dataset_id: str = Path(..., title="数据集ID"),
    session: AsyncSession = Depends(get_async_session)
):
    """检查数据集结构变更
    
    Args:
        dataset_id: 数据集ID
    """
    try:
        # 查询数据集
        result = await session.execute(
            select(DataSet).filter(DataSet.id == dataset_id)
        )
        dataset = result.scalar_one_or_none()
        
        if not dataset:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"数据集不存在: {dataset_id}"
            )
        
        # 检查结构变更
        changes = await check_structure_changes(dataset, session)
        
        # 如果有变更，更新数据集状态
        if changes["has_changes"]:
            dataset.sync_status = "structure_changed"
            dataset.last_structure_changed_at = datetime.now()
            session.add(dataset)
            await session.commit()
        
        # 构建响应
        response = {
            "success": True,
            "has_changes": changes["has_changes"],
            "message": "检测到结构变更" if changes["has_changes"] else "未检测到结构变更",
            "changes": {
                "added": [{"name": item["field_name"]} for item in changes["changes"]["added"]],
                "removed": [{"name": item["field_name"]} for item in changes["changes"]["removed"]],
                "modified": [{
                    "name": item["field_name"],
                    "changes": item["changes"]
                } for item in changes["changes"]["modified"]]
            }
        }
        
        return response
    
    except HTTPException:
        raise
    
    except Exception as e:
        await session.rollback()
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"检查结构变更失败: {str(e)}"
        )


@router.post("/{dataset_id}/apply-changes", response_model=Dict[str, Any])
async def apply_changes(
    dataset_id: str = Path(..., title="数据集ID"),
    session: AsyncSession = Depends(get_async_session)
):
    """应用数据集结构变更
    
    Args:
        dataset_id: 数据集ID
    """
    try:
        # 查询数据集
        result = await session.execute(
            select(DataSet).filter(DataSet.id == dataset_id)
        )
        dataset = result.scalar_one_or_none()
        
        if not dataset:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"数据集不存在: {dataset_id}"
            )
        
        # 应用结构变更
        result = await apply_structure_changes(dataset, session)
        
        # 保存更改
        await session.commit()
        
        return {
            "success": True,
            "message": result["message"],
            "applied_changes": result["applied_changes"]
        }
    
    except HTTPException:
        raise
    
    except Exception as e:
        await session.rollback()
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"应用结构变更失败: {str(e)}"
        )


@router.post("/{dataset_id}/sync", response_model=Dict[str, Any])
async def sync_data(
    dataset_id: str = Path(..., title="数据集ID"),
    background_tasks: BackgroundTasks = None,
    session: AsyncSession = Depends(get_async_session)
):
    """同步数据集数据
    
    Args:
        dataset_id: 数据集ID
        background_tasks: 后台任务
    """
    try:
        # 查询数据集
        result = await session.execute(
            select(DataSet).filter(DataSet.id == dataset_id)
        )
        dataset = result.scalar_one_or_none()
        
        if not dataset:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"数据集不存在: {dataset_id}"
            )
        
        # 检查数据集状态
        if dataset.sync_status == "syncing":
            return {
                "success": False,
                "message": "数据集正在同步中，请稍后再试"
            }
        
        # 如果有结构变更且策略为通知，则返回错误
        if dataset.sync_status == "structure_changed" and dataset.structure_change_strategy == "notify":
            return {
                "success": False,
                "message": "检测到结构变更，请先应用变更后再同步"
            }
        
        # 开始同步（后台任务或同步执行）
        if background_tasks:
            # 更新数据集状态为同步中
            dataset.sync_status = "syncing"
            session.add(dataset)
            await session.commit()
            
            # 添加后台任务
            background_tasks.add_task(sync_dataset, dataset, session)
            
            return {
                "success": True,
                "message": "数据集同步已启动，请稍后查看同步状态"
            }
        else:
            # 同步执行
            result = await sync_dataset(dataset, session)
            
            # 保存更改
            await session.commit()
            
            return result
    
    except HTTPException:
        raise
    
    except Exception as e:
        await session.rollback()
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"同步数据集失败: {str(e)}"
        )


# ==================== 字段管理 ====================

@router.get("/{dataset_id}/fields", response_model=Dict[str, Any])
async def get_fields(
    dataset_id: str = Path(..., title="数据集ID"),
    session: AsyncSession = Depends(get_async_session)
):
    """获取数据集字段列表
    
    Args:
        dataset_id: 数据集ID
    """
    try:
        # 检查数据集是否存在
        result = await session.execute(
            select(DataSet).filter(DataSet.id == dataset_id)
        )
        dataset = result.scalar_one_or_none()
        
        if not dataset:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"数据集不存在: {dataset_id}"
            )
        
        # 获取字段列表
        fields_result = await session.execute(
            select(DataField).filter(
                DataField.dataset_id == dataset_id
            ).order_by(DataField.ordinal_position)
        )
        fields = fields_result.scalars().all()
        
        return {
            "success": True,
            "data": [field.dict() for field in fields]
        }
    
    except HTTPException:
        raise
    
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"获取字段列表失败: {str(e)}"
        )


@router.get("/{dataset_id}/fields/{field_id}", response_model=Dict[str, Any])
async def get_field(
    dataset_id: str = Path(..., title="数据集ID"),
    field_id: str = Path(..., title="字段ID"),
    session: AsyncSession = Depends(get_async_session)
):
    """获取字段详情
    
    Args:
        dataset_id: 数据集ID
        field_id: 字段ID
    """
    try:
        # 查询字段
        result = await session.execute(
            select(DataField).filter(
                DataField.id == field_id,
                DataField.dataset_id == dataset_id
            )
        )
        field = result.scalar_one_or_none()
        
        if not field:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"字段不存在: {field_id}"
            )
        
        return {
            "success": True,
            "data": field.dict()
        }
    
    except HTTPException:
        raise
    
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"获取字段详情失败: {str(e)}"
        )


@router.put("/{dataset_id}/fields/{field_id}", response_model=Dict[str, Any])
async def update_field(
    field_data: Dict[str, Any],
    dataset_id: str = Path(..., title="数据集ID"),
    field_id: str = Path(..., title="字段ID"),
    session: AsyncSession = Depends(get_async_session)
):
    """更新字段元数据
    
    Args:
        dataset_id: 数据集ID
        field_id: 字段ID
        field_data: 字段数据
    """
    try:
        # 查询字段
        result = await session.execute(
            select(DataField).filter(
                DataField.id == field_id,
                DataField.dataset_id == dataset_id
            )
        )
        field = result.scalar_one_or_none()
        
        if not field:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"字段不存在: {field_id}"
            )
        
        # 查询数据集
        dataset_result = await session.execute(
            select(DataSet).filter(DataSet.id == dataset_id)
        )
        dataset = dataset_result.scalar_one_or_none()
        
        if not dataset:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"数据集不存在: {dataset_id}"
            )
        
        # 更新字段元数据
        updated_field = await update_field_metadata(field, field_data, session)
        
        # 重新生成数据集结构信息
        dataset_structure = await DataHelp.get_dataset_structure(dataset_id, session)
        dataset.dataset_summary = DataHelp.to_json(dataset_structure)
        session.add(dataset)
        
        # 保存更改
        await session.commit()
        await session.refresh(updated_field)
        
        return {
            "success": True,
            "message": "字段元数据更新成功",
            "data": updated_field.to_dict()
        }
    
    except HTTPException:
        raise
    
    except Exception as e:
        await session.rollback()
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"更新字段元数据失败: {str(e)}"
        )


@router.get("/{dataset_id}/fields/{field_id}/suggestions", response_model=Dict[str, Any])
async def get_field_suggestions(
    dataset_id: str = Path(..., title="数据集ID"),
    field_id: str = Path(..., title="字段ID"),
    session: AsyncSession = Depends(get_async_session)
):
    """获取字段元数据建议
    
    Args:
        dataset_id: 数据集ID
        field_id: 字段ID
    """
    try:
        # 查询字段
        result = await session.execute(
            select(DataField).filter(
                DataField.id == field_id,
                DataField.dataset_id == dataset_id
            )
        )
        field = result.scalar_one_or_none()
        
        if not field:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"字段不存在: {field_id}"
            )
        
        # 生成元数据建议
        suggestions = await generate_metadata_suggestions(field, session)
        
        return {
            "success": True,
            "data": suggestions
        }
    
    except HTTPException:
        raise
    
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"获取字段元数据建议失败: {str(e)}"
        )


@router.get("/{dataset_id}/relationships", response_model=Dict[str, Any])
async def get_relationships(
    dataset_id: str = Path(..., title="数据集ID"),
    session: AsyncSession = Depends(get_async_session)
):
    """获取数据集字段关系
    
    Args:
        dataset_id: 数据集ID
    """
    try:
        # 检查数据集是否存在
        result = await session.execute(
            select(DataSet).filter(DataSet.id == dataset_id)
        )
        dataset = result.scalar_one_or_none()
        
        if not dataset:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"数据集不存在: {dataset_id}"
            )
        
        # 获取字段关系
        relationships = await get_field_relationship(dataset_id, session)
        
        return {
            "success": True,
            "data": relationships
        }
    
    except HTTPException:
        raise
    
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"获取字段关系失败: {str(e)}"
        )


# ==================== 结构变更管理 ====================

@router.get("/{dataset_id}/changes", response_model=Dict[str, Any])
async def get_changes(
    dataset_id: str = Path(..., title="数据集ID"),
    status: Optional[str] = Query(None, description="变更状态：pending, approved, rejected, applied"),
    session: AsyncSession = Depends(get_async_session)
):
    """获取数据集结构变更列表
    
    Args:
        dataset_id: 数据集ID
        status: 变更状态
    """
    try:
        # 检查数据集是否存在
        result = await session.execute(
            select(DataSet).filter(DataSet.id == dataset_id)
        )
        dataset = result.scalar_one_or_none()
        
        if not dataset:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"数据集不存在: {dataset_id}"
            )
        
        # 构建查询
        query = select(FieldChangeLog).filter(
            FieldChangeLog.dataset_id == dataset_id
        ).order_by(FieldChangeLog.created_at.desc())
        
        # 应用状态过滤
        if status:
            query = query.filter(FieldChangeLog.status == status)
        
        # 执行查询
        result = await session.execute(query)
        changes = result.scalars().all()
        
        return {
            "success": True,
            "data": [change.to_dict() for change in changes]
        }
    
    except HTTPException:
        raise
    
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"获取结构变更列表失败: {str(e)}"
        )


@router.put("/{dataset_id}/changes/{change_id}/status", response_model=Dict[str, Any])
async def update_change_status(
    status_data: Dict[str, Any],
    dataset_id: str = Path(..., title="数据集ID"),
    change_id: str = Path(..., title="变更ID"),
    session: AsyncSession = Depends(get_async_session)
):
    """更新结构变更状态
    
    Args:
        dataset_id: 数据集ID
        change_id: 变更ID
        status_data: 包含状态的数据
    """
    try:
        # 检查变更是否存在
        result = await session.execute(
            select(FieldChangeLog).filter(
                FieldChangeLog.id == change_id,
                FieldChangeLog.dataset_id == dataset_id
            )
        )
        change = result.scalar_one_or_none()
        
        if not change:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"结构变更不存在: {change_id}"
            )
        
        # 更新状态
        new_status = status_data.get("status")
        if not new_status or new_status not in ["pending", "approved", "rejected", "applied"]:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="无效的状态值"
            )
        
        change.status = new_status
        
        if new_status == "applied":
            change.applied_at = datetime.now()
        
        # 保存更改
        session.add(change)
        await session.commit()
        await session.refresh(change)
        
        return {
            "success": True,
            "message": "结构变更状态更新成功",
            "data": change.to_dict()
        }
    
    except HTTPException:
        raise
    
    except Exception as e:
        await session.rollback()
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"更新结构变更状态失败: {str(e)}"
        )


@router.post("/{dataset_id}/sync-fields", response_model=Dict[str, Any])
async def sync_fields(
    dataset_id: str = Path(..., title="数据集ID"),
    session: AsyncSession = Depends(get_async_session)
):
    """同步数据源字段到数据集
    
    将数据源中的表字段信息同步到数据集的DataField表中。
    已存在的字段将保留其显示名称、描述等用户自定义信息，仅更新技术性元数据。
    不再存在的字段将标记为不可见，但不会删除。
    
    Args:
        dataset_id: 数据集ID
    """
    try:
        # 查询数据集
        result = await session.execute(
            select(DataSet).filter(DataSet.id == dataset_id)
        )
        dataset = result.scalar_one_or_none()
        
        if not dataset:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"数据集不存在: {dataset_id}"
            )
        
        # 执行字段同步
        sync_result = await sync_datasource_fields_to_dataset(
            data_source_id=dataset.data_source_id,
            dataset_id=dataset_id,
            session=session
        )
        
        # 提交事务
        await session.commit()
        
        # 更新数据集结构
        dataset_structure = await DataHelp.get_dataset_structure(dataset_id, session)
        dataset.dataset_summary = DataHelp.to_json(dataset_structure)
        await session.commit()
        
        return {
            "success": True,
            "message": sync_result["message"],
            "data": {
                "fields_created": sync_result["fields_created"],
                "fields_kept": sync_result["fields_kept"],
                "fields_total": sync_result["fields_total"]
            }
        }
    
    except HTTPException:
        raise
    
    except Exception as e:
        # 发生异常时回滚事务
        await session.rollback()
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"同步数据源字段到数据集失败: {str(e)}"
        )


# ==================== 管理员接口 ====================

@router.get("/admin/messages", response_model=Dict[str, Any])
async def admin_get_all_messages(
    start_date: Optional[str] = Query(None, description="开始日期 (YYYY-MM-DD)"),
    end_date: Optional[str] = Query(None, description="结束日期 (YYYY-MM-DD)"),
    user_id: Optional[str] = Query(None, description="用户ID"),
    session_id: Optional[str] = Query(None, description="会话ID"),
    page: int = Query(1, ge=1, description="页码"),
    page_size: int = Query(20, ge=1, le=100, description="每页记录数"),
    current_user: User = Depends(current_active_user),
    session: AsyncSession = Depends(get_async_session)
):
    """管理员获取所有用户消息和系统回复
    
    此接口允许管理员查看系统中的所有用户消息及其对应的系统回复，支持按日期范围、用户ID和会话ID筛选。
    
    Args:
        start_date: 开始日期 (YYYY-MM-DD)
        end_date: 结束日期 (YYYY-MM-DD)
        user_id: 用户ID
        session_id: 会话ID
        page: 页码
        page_size: 每页记录数
    """
    try:
       
        
        # 构建查询 - 只查询用户消息（message_type='user'）
        query = select(UserMsg).filter(UserMsg.message_type == 'user').order_by(UserMsg.timestamp.desc())
        
        # 应用日期过滤
        if start_date:
            start_datetime = datetime.strptime(f"{start_date} 00:00:00", "%Y-%m-%d %H:%M:%S")
            query = query.filter(UserMsg.timestamp >= start_datetime)
        
        if end_date:
            end_datetime = datetime.strptime(f"{end_date} 23:59:59", "%Y-%m-%d %H:%M:%S")
            query = query.filter(UserMsg.timestamp <= end_datetime)
        
        # 应用用户ID过滤
        if user_id:
            query = query.filter(UserMsg.user_id == user_id)
        
        # 应用会话ID过滤
        if session_id:
            query = query.filter(UserMsg.session_id == session_id)
        
        # 计算总数
        count_query = select(func.count()).select_from(query.subquery())
        total_count = await session.scalar(count_query)
        
        # 分页查询
        offset = (page - 1) * page_size
        query = query.offset(offset).limit(page_size)
        
        # 执行查询
        result = await session.execute(query)
        user_messages = result.scalars().all()
        
        # 获取用户消息ID列表，用于查询系统回复
        user_msg_ids = [str(msg.id) for msg in user_messages]
        
        # 查询系统回复消息
        if user_msg_ids:
            system_msg_query = select(UserMsg).filter(
                UserMsg.message_type == 'system',
                UserMsg.message_sourceid.in_(user_msg_ids)
            )
            system_msg_result = await session.execute(system_msg_query)
            system_messages = system_msg_result.scalars().all()
            
            # 创建系统消息映射，键为message_sourceid（用户消息ID）
            system_msg_map = {}
            for sys_msg in system_messages:
                source_id = sys_msg.message_sourceid
                if source_id not in system_msg_map:
                    system_msg_map[source_id] = []
                system_msg_map[source_id].append(sys_msg)
        else:
            system_msg_map = {}
        
        # 获取相关会话和用户信息
        response_data = []
        for msg in user_messages:
            # 使用dict()方法获取字典表示
            if hasattr(msg, "dict") and callable(getattr(msg, "dict")):
                msg_dict = msg.dict()
            else:
                msg_dict = {
                    "id": str(msg.id),
                    "user_id": str(msg.user_id) if msg.user_id else None,
                    "session_id": str(msg.session_id) if msg.session_id else None,
                    "message_type": msg.message_type,
                    "message": msg.message,
                    "message_model": getattr(msg, "message_model", None),
                    "message_sql": getattr(msg, "message_sql", None),
                    "message_intent": getattr(msg, "message_intent", None),
                    "message_confirm": getattr(msg, "message_confirm", None),
                    "message_suggest": getattr(msg, "message_suggest", None),
                    "chart_option": getattr(msg, "chart_option", None),
                    "timestamp": msg.timestamp.isoformat() if msg.timestamp else None
                }
            
            # 查询会话信息
            if msg.session_id:
                session_result = await session.execute(
                    select(UserSession).filter(UserSession.id == msg.session_id)
                )
                user_session = session_result.scalar_one_or_none()
                msg_dict["session"] = user_session
            
            # 查询用户信息
            if msg.user_id:
                user_result = await session.execute(
                    select(User).filter(User.id == msg.user_id)
                )
                user = user_result.scalar_one_or_none()
                if user:
                    msg_dict["user"] = {
                        "id": str(user.id),
                        "email": user.email,
                        "username": getattr(user, "username", None),
                        "display_name": getattr(user, "nickname", None)
                    }
            
            # 尝试解析消息内容
            try:
                if msg.message and isinstance(msg.message, str):
                    msg_dict["message_json"] = json.loads(msg.message)
            except Exception:
                # 如果不是有效的JSON，保持原始内容
                pass
            
            # 添加系统回复消息
            msg_id = str(msg.id)
            if msg_id in system_msg_map:
                system_replies = []
                for sys_msg in system_msg_map[msg_id]:
                    if hasattr(sys_msg, "dict") and callable(getattr(sys_msg, "dict")):
                        sys_dict = sys_msg.dict()
                    else:
                        sys_dict = {
                            "id": str(sys_msg.id),
                            "session_id": str(sys_msg.session_id) if sys_msg.session_id else None,
                            "message_type": sys_msg.message_type,
                            "message": sys_msg.message,
                            "message_model": getattr(sys_msg, "message_model", None),
                            "message_sql": getattr(sys_msg, "message_sql", None),
                            "message_intent": getattr(sys_msg, "message_intent", None),
                            "message_confirm": getattr(sys_msg, "message_confirm", None),
                            "message_suggest": getattr(sys_msg, "message_suggest", None),
                            "chart_option": getattr(sys_msg, "chart_option", None),
                            "timestamp": sys_msg.timestamp.isoformat() if sys_msg.timestamp else None
                        }
                    
                    # 尝试解析系统回复内容
                    try:
                        if sys_msg.message and isinstance(sys_msg.message, str):
                            sys_dict["message_json"] = json.loads(sys_msg.message)
                    except Exception:
                        pass
                    
                    system_replies.append(sys_dict)
                
                msg_dict["system_replies"] = system_replies
            
            response_data.append(msg_dict)
        
        return {
            "success": True,
            "data": response_data,
            "total": total_count,
            "page": page,
            "page_size": page_size,
            "total_pages": (total_count + page_size - 1) // page_size
        }
    
    except HTTPException:
        raise
    
    except Exception as e:
        logging.error(f"获取用户消息失败: {str(e)}", exc_info=True)
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"获取用户消息失败: {str(e)}"
        ) 